package io.jjz.config;

import org.apache.commons.io.FileUtils;
import org.apache.commons.io.IOUtils;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.ConnectionFactory;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.io.File;
import java.io.IOException;
import java.io.InputStream;

@Configuration
public class HBaseConfig {

    private static final Logger LOGGER = LoggerFactory.getLogger(HBaseConfig.class);

    public static final String ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME = "Client";
    public static final String ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL = "zookeeper/hadoop";
    public static String userPrincipal = "CORPLU25JHSK";

    public static String userKeytabPath;
    public static String krb5Path;

    //@PostConstruct
    public void preLogin() {
        try {
            // 以流的方式读取
            InputStream keyInputStream = EnvConfig.class.getClassLoader().getResourceAsStream("user.keytab");
            InputStream krb5InputStream = EnvConfig.class.getClassLoader().getResourceAsStream("krb5.conf");

            // 创建临时文件
            File keyFile = File.createTempFile("user", "keytab");
            File krb5File = File.createTempFile("krb5", "conf");

            //将流的内容复制到文件上
            FileUtils.copyInputStreamToFile(keyInputStream, keyFile);
            FileUtils.copyInputStreamToFile(krb5InputStream, krb5File);

            IOUtils.closeQuietly(keyInputStream);
            IOUtils.closeQuietly(krb5InputStream);

            userKeytabPath = keyFile.getAbsolutePath();
            krb5Path = krb5File.getAbsolutePath();

            System.setProperty("zookeeper.server.principal", ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL); //LoginUtil.setZookeeperServerPrincipal
            System.setProperty("java.security.krb5.conf", krb5Path); //LoginUtil.login
            System.setProperty("zookeeper.sasl.client", "true");

        } catch (IOException e) {
            LOGGER.error("preLogin fail....");
            throw new RuntimeException(e);
        }
    }

    /**
     * 加载集群配置信息
     */
    @Bean(name = "configuration")
    public org.apache.hadoop.conf.Configuration getHBaseConfiguration() {
        org.apache.hadoop.conf.Configuration conf = HBaseConfiguration.create();
        //conf.addResource(new Path("core-site.xml"));
        //conf.addResource(new Path("hdfs-site.xml"));
        conf.addResource(new Path("hbase-site.xml"));
        return conf;
    }


    @Bean(name = "hBaseConnection", destroyMethod = "close")
    //@Scope(value = ConfigurableBeanFactory.SCOPE_PROTOTYPE, proxyMode = ScopedProxyMode.TARGET_CLASS)
    public Connection getConnection(@Qualifier("configuration") org.apache.hadoop.conf.Configuration conf) throws IOException {
        //if (User.isHBaseSecurityEnabled(conf)) {}
        LOGGER.info("创建hbase连接");
        //LoginUtil.setJaasConf(Constants.ZOOKEEPER_DEFAULT_LOGIN_CONTEXT_NAME, Constants.userPrincipal, Constants.userKeytabPath);
        //LoginUtil.setZookeeperServerPrincipal(Constants.ZOOKEEPER_DEFAULT_SERVER_PRINCIPAL);
        //LoginUtil.login(Constants.userPrincipal, Constants.userKeytabPath, Constants.krb5Path, conf);

        //UserGroupInformation.setConfiguration(conf);
        //UserGroupInformation.loginUserFromKeytab(Constants.userPrincipal, Constants.userKeytabPath);

        //UserProvider provider = UserProvider.instantiate(conf);
        //User user = provider.create(ugi);
        /*return UserGroupInformation.getLoginUser().doAs((PrivilegedExceptionAction<Connection>) () -> {
            try {
                return ConnectionFactory.createConnection(conf);
            } catch (IOException e) {
                e.printStackTrace();
            }
            return null;
        });*/
        return ConnectionFactory.createConnection(conf);
    }

}
